package com.chukun.flink.stream.window.process.windows;

import com.chukun.flink.stream.window.source.SourceForWindow;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.evictors.CountEvictor;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.util.ArrayList;
import java.util.List;

/**
 * @author chukun
 * @version 1.0.0
 * @description 窗口剔除器的处理逻辑
 * @createTime 2022年05月24日 22:20:00
 */
public class TumblingWindowEvitorOperator {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 自定义数据源
        DataStreamSource<Tuple3<String, Integer, String>> streamSource = env.addSource(new SourceForWindow(1000, false));

        // map转化操作
        SingleOutputStreamOperator<Tuple2<String, List<Integer>>> mapStream = streamSource.map(new MapFunction<Tuple3<String, Integer, String>, Tuple2<String, List<Integer>>>() {
            @Override
            public Tuple2<String, List<Integer>> map(Tuple3<String, Integer, String> tuple) throws Exception {
                List<Integer> list = new ArrayList<>();
                list.add(tuple.f1);
                return Tuple2.of(tuple.f0, list);
            }
        });

        // 根据 数据流中的 f0 字段分组
        KeyedStream<Tuple2<String, List<Integer>>, String> keyedStream = mapStream.keyBy((key) -> key.f0);


        SingleOutputStreamOperator<Tuple2<String, List<Integer>>> reduceStream = keyedStream
                // 滚动窗口，窗口大小5秒
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                // 指定内置的CountEvictor剔除器，使得窗口内最多保留3个元素用于窗口计算
                .evictor(CountEvictor.of(3))
                // 对窗口应用ReduceFunction函数
                .reduce(new ReduceFunction<Tuple2<String, List<Integer>>>() {
                    @Override
                    public Tuple2<String, List<Integer>> reduce(Tuple2<String, List<Integer>> value01, Tuple2<String, List<Integer>> value02) throws Exception {
                        value01.f1.add(value02.f1.get(0));
                        return value01;
                    }
                });

        SingleOutputStreamOperator<Tuple2<String, List<Integer>>> copyedReduceStream = keyedStream
                // 滚动窗口，窗口大小5秒
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                // 对窗口应用ReduceFunction函数
                .reduce(new ReduceFunction<Tuple2<String, List<Integer>>>() {
                    @Override
                    public Tuple2<String, List<Integer>> reduce(Tuple2<String, List<Integer>> value01, Tuple2<String, List<Integer>> value02) throws Exception {
                        value01.f1.add(value02.f1.get(0));
                        return value01;
                    }
                });

        reduceStream.print("使用元素剔除器的窗口计算");

        copyedReduceStream.print("未使用元素剔除器的窗口计算");

        env.execute("TumblingWindowEvitorOperator");

    }
}
